<?php
namespace app\service;

use think\Db;
use think\facade\Log;

/**
 * MQTT消息处理类
 * 专门用于处理接收到的MQTT消息并保存到数据库
 */
class MqttMessageHandler
{
    /**
     * 消息批处理大小
     * 当消息达到此数量时批量插入数据库
     */
    protected $batchSize = 50;
    
    /**
     * 消息缓冲区
     */
    protected $messageBuffer = [];
    
    /**
     * 上次保存时间
     */
    protected $lastSaveTime = 0;
    
    /**
     * 最大缓冲时间(秒)
     * 即使未达到批处理大小，超过此时间也会保存
     */
    protected $maxBufferTime = 10;
    
    /**
     * 处理消息
     * @param array $message 消息数据
     * @return bool 处理结果
     */
    public function handle(array $message): bool
    {
        try {
            // 解析消息
            $topic = $message['topic'] ?? '';
            $payload = $message['message'] ?? '';
            $qos = $message['qos'] ?? 0;
            $time = time();
            
            // 记录日志
            Log::info("MQTT消息处理: 主题={$topic}, 内容={$payload}, QoS={$qos}");
            
            // 根据主题分类处理
            $data = $this->parseMessage($topic, $payload);
            if (empty($data)) {
                Log::error("MQTT消息解析失败: " . json_encode($message, JSON_UNESCAPED_UNICODE));
                return false;
            }
            
            // 添加时间戳
            $data['create_time'] = $time;
            $data['topic'] = $topic;
            $data['qos'] = $qos;
            
            // 添加到缓冲区
            $this->addToBuffer($data);
            
            return true;
        } catch (\Throwable $e) {
            Log::error('MQTT消息处理异常: ' . $e->getMessage() . "\n" . $e->getTraceAsString());
            return false;
        }
    }
    
    /**
     * 解析消息内容
     * @param string $topic 主题
     * @param string $payload 消息内容
     * @return array 解析后的数据
     */
    protected function parseMessage(string $topic, string $payload): array
    {
        Log::info("开始解析消息: 主题={$topic}, 内容={$payload}");
        
        // 初始化数据数组，只包含数据库中存在的字段
        $data = [
            'raw_message' => $payload
        ];
        
        // 尝试解析JSON
        $jsonData = json_decode($payload, true);
        if (json_last_error() !== JSON_ERROR_NONE) {
            // 非JSON格式，已经作为普通文本保存在raw_message中
            Log::info("消息不是有效的JSON格式: " . json_last_error_msg());
        } else {
            Log::info("成功解析JSON消息: " . json_encode($jsonData, JSON_UNESCAPED_UNICODE));
            // 将解析后的JSON保存为字符串，而不是直接使用键值对
            $data['raw_message'] = $payload;
        }
        
        // 根据主题分类处理
        if (strpos($topic, 'testtopic/') === 0) {
            // 测试主题
            $data['message_type'] = 'test';
            Log::info("识别为测试主题消息");
        } elseif (strpos($topic, 'sensor/') === 0) {
            // 传感器数据
            $data['message_type'] = 'sensor';
            // 从主题中提取传感器ID
            $parts = explode('/', $topic);
            $data['sensor_id'] = $parts[1] ?? '';
        } elseif (strpos($topic, 'device/') === 0) {
            // 设备数据
            $data['message_type'] = 'device';
            // 从主题中提取设备ID
            $parts = explode('/', $topic);
            $data['device_id'] = $parts[1] ?? '';
        } else {
            // 其他主题
            $data['message_type'] = 'other';
        }
        
        Log::info("消息解析结果: " . json_encode($data, JSON_UNESCAPED_UNICODE));
        return $data;
    }
    
    /**
     * 添加消息到缓冲区
     * @param array $data 消息数据
     */
    protected function addToBuffer(array $data): void
    {
        // 添加到缓冲区
        $this->messageBuffer[] = $data;
        Log::info("消息已添加到缓冲区, 当前缓冲区大小: " . count($this->messageBuffer));
        
        // 检查是否需要保存
        $currentTime = time();
        $bufferFull = count($this->messageBuffer) >= $this->batchSize;
        $timeExceeded = ($currentTime - $this->lastSaveTime) >= $this->maxBufferTime && !empty($this->messageBuffer);
        
        if ($bufferFull) {
            Log::info("缓冲区已满，执行保存");
            $this->saveToDatabase();
            $this->lastSaveTime = $currentTime;
        } elseif ($timeExceeded) {
            Log::info("缓冲时间已超过最大值，执行保存");
            $this->saveToDatabase();
            $this->lastSaveTime = $currentTime;
        } else {
            Log::info("暂不保存，等待更多消息或超时");
        }
    }
    
    /**
     * 保存消息到数据库
     */
    protected function saveToDatabase(): void
    {
        if (empty($this->messageBuffer)) {
            Log::info("缓冲区为空，无需保存");
            return;
        }
        
        try {
            Log::info("开始保存消息到数据库, 消息数量: " . count($this->messageBuffer));
            
            // 检查数据库连接
            try {
                Db::query("SELECT 1");
                Log::info("数据库连接正常");
            } catch (\Throwable $e) {
                Log::error("数据库连接测试失败: " . $e->getMessage());
            }
            
            // 批量插入数据库
            $result = Db::name('mqtt_messages')->insertAll($this->messageBuffer);
            Log::info("数据库插入结果: " . ($result !== false ? "成功, 影响行数: {$result}" : "失败"));
            
            // 清空缓冲区
            $this->messageBuffer = [];
            Log::info("缓冲区已清空");
        } catch (\Throwable $e) {
            Log::error('保存MQTT消息到数据库失败: ' . $e->getMessage() . "\n" . $e->getTraceAsString());
            
            // 失败后，保留一部分消息重试
            // 避免缓冲区无限增长
            if (count($this->messageBuffer) > $this->batchSize * 2) {
                $this->messageBuffer = array_slice($this->messageBuffer, -$this->batchSize);
                Log::info("缓冲区已裁剪，当前大小: " . count($this->messageBuffer));
            }
        }
    }
    
    /**
     * 强制保存所有缓冲的消息
     */
    public function flush(): void
    {
        Log::info("强制保存所有缓冲消息");
        $this->saveToDatabase();
    }
    
    /**
     * 析构函数，确保所有消息都被保存
     */
    public function __destruct()
    {
        Log::info("MqttMessageHandler析构，保存剩余消息");
        $this->flush();
    }
} 